package Consumer

import (
	"context"
	"deepgox-go-admin/Infrastructures/Rom"
	"deepgox-go-admin/Models/Manager"
	"deepgox-go-admin/Settings"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"log"
)

// RConsumer 声明一个全局的RConsumer变量
var RConsumer *rocketmq.PushConsumer

func Init(cfg *Settings.RocketmqConfig) (err error) {
	RConsumer, _ := rocketmq.NewPushConsumer(
		consumer.WithNameServer(cfg.NameServer),
		consumer.WithGroupName("deepgox"),
	)
	err = RConsumer.Subscribe("manager_topic", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			// 这里判断topic 的类型
			// todo 功能拓展
			fmt.Printf("接收到消息: %s\n", msg.Topic)
			// 获取消息的 body 数据
			body := string(msg.Body)
			var manager Manager.AddModelHandler
			err := json.Unmarshal(msg.Body, &manager)
			if err != nil {
				return 0, err
			}
			result := Rom.Db.Exec("INSERT into manager(username,`password`,email,gender) VALUES(?, ?, ? , ? )", manager.UserName, manager.Password, manager.Email, manager.Gender)
			if result.RowsAffected != 1 {
				return 0, result.Error
			}
			fmt.Printf("接收到消息: %s\n", body)
		}
		return consumer.ConsumeSuccess, nil
	})
	// 启动消费者
	if err := RConsumer.Start(); err != nil {
		log.Fatal("Consumer Start Fail, ", err)
	}

	return nil
}
